package beautiful.butterfly.drds.data_exchange.taskgroup.runner;

import beautiful.butterfly.drds.data_exchange.plugin.AbstractTaskPlugin;
import beautiful.butterfly.drds.data_exchange.spi.AbstractReader;
import beautiful.butterfly.drds.data_exchange.statistics.PerfRecord;
import beautiful.butterfly.drds.data_exchange.statistics.Phase;
import beautiful.butterfly.drds.data_exchange.statistics.communication.CommunicationTool;
import beautiful.butterfly.drds.data_exchange.transport.exchanger.Produce;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 单个slice的reader执行调用
 */
public class ReaderRunner extends AbstractRunner implements Runnable
{

    private static final Logger LOG = LoggerFactory
            .getLogger(ReaderRunner.class);

    private Produce produce;

    public ReaderRunner(AbstractTaskPlugin abstractTaskPlugin)
    {
        super(abstractTaskPlugin);
    }

    public void setProduce(Produce produce)
    {
        this.produce = produce;
    }

    public void run()
    {
        assert null != this.produce;

        AbstractReader.AbstractTask abstractTaskReader = (AbstractReader.AbstractTask) this.getPlugin();

        //统计waitWriterTime，并且在finally才end。
        PerfRecord channelWaitWrite = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.wait_write_time);
        try
        {
            channelWaitWrite.start();

            LOG.debug("task reader starts to do init ...");
            PerfRecord initPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.read_task_init);
            initPerfRecord.start();
            abstractTaskReader.init();
            initPerfRecord.end();

            LOG.debug("task reader starts to do prepare ...");
            PerfRecord preparePerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.read_task_prepare);
            preparePerfRecord.start();
            abstractTaskReader.prepare();
            preparePerfRecord.end();

            LOG.debug("task reader starts to read ...");
            PerfRecord dataPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.read_task_data);
            dataPerfRecord.start();
            abstractTaskReader.startRead(produce);
            produce.terminate();

            dataPerfRecord.addCount(CommunicationTool.getTotalReadRecords(super.getRunnerCommunication()));
            dataPerfRecord.addSize(CommunicationTool.getTotalReadBytes(super.getRunnerCommunication()));
            dataPerfRecord.end();

            LOG.debug("task reader starts to do post ...");
            PerfRecord postPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.read_task_post);
            postPerfRecord.start();
            abstractTaskReader.post();
            postPerfRecord.end();
            // automatic pushAllToRecordQueue
            // super.markSuccess(); 这里不能标记为成功，成功的标志由 writerRunner 来标志（否则可能导致 reader 先结束，而 writer 还没有结束的严重 bug）
        } catch (Throwable e)
        {
            LOG.error("AbstractReader runner Received Exceptions:", e);
            super.markFail(e);
        } finally
        {
            LOG.debug("task reader starts to do destroy ...");
            PerfRecord desPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.read_task_destroy);
            desPerfRecord.start();
            super.destroy();
            desPerfRecord.end();

            channelWaitWrite.end(super.getRunnerCommunication().getLongCounter(CommunicationTool.wait_writer_time));

            long transformerUsedTime = super.getRunnerCommunication().getLongCounter(CommunicationTool.transformer_used_time);
            if (transformerUsedTime > 0)
            {
                PerfRecord transformerRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.transformer_time);
                transformerRecord.start();
                transformerRecord.end(transformerUsedTime);
            }
        }
    }

    public void shutdown()
    {
        produce.shutdown();
    }
}
